import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.channels.*
import kotlinx.coroutines.experimental.selects.select
import kotlinx.coroutines.experimental.sync.Mutex
import kotlinx.coroutines.experimental.sync.withLock
import java.util.*
import java.util.concurrent.atomic.AtomicInteger
import kotlin.coroutines.experimental.CoroutineContext
import kotlin.coroutines.experimental.coroutineContext
import kotlin.system.measureTimeMillis


data class Ball(var hits: Int)

sealed class CounterMsg
object IncCounter : CounterMsg()
class GetCounter(val response: CompletableDeferred<Int>) : CounterMsg()

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main(args: Array<String>) {
    do38()
}

fun do38() {
    runBlocking {
        println("do38()")
        val list = asyncStringsList()
        val result = select<String> {
            list.withIndex().forEach { (index, deferred)
                ->
                deferred.onAwait { answer ->
                    "Deferred $index produce answer '$answer'"
                }
            }
        }
        println(result)
        val cntActive = list.count { it.isActive }
        println("$cntActive coroutines are still active")


        list.withIndex().forEach {
            (index, deferred)->
            println("index=$index, value=${deferred.await()}")
        }


        println("Done!")
    }//end runBlocking
}

fun asyncStringsList(): List<Deferred<String>> {
    val random = Random(3)
    return List(12)
    {
        asyncString(random.nextInt(1000))
    }
}

fun asyncString(time: Int) = async {
    delay(time.toLong())
    "Waited for $time ms"
}

fun do37() {
    runBlocking {
        println("do37()")

        val side = Channel<Int>()
        launch(coroutineContext) {
            for (y in side) {
                println("Side channel has $y")
            }
        }//end launch

        val receiveChannel = produceNumbers(coroutineContext, side)
        for (item in receiveChannel) {
            println("Consuming $item")
            delay(250L)
        }

        println("Done consuming")
        coroutineContext.cancelChildren()
    }
}

fun produceNumbers(context: CoroutineContext, side: SendChannel<Int>) = produce<Int>(context) {
    for (num in 1..20) {
        delay(100L)
        select<Unit> {
            onSend(num) {}
            side.onSend(num) {}
        }
    }
}


fun do36() {
    runBlocking<Unit> {
        val a = produce<String>(coroutineContext) {
            repeat(4) {
                //                send("Hello $it")
            }
        }

        val b = produce<String>(coroutineContext) {
            repeat(4) {
                //                send("World $it")
            }
        }

        repeat(1) {
            println(selectAorB(a, b))
        }
        coroutineContext.cancelChildren()
        println("Done!")
    }
}

suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
        select<String> {
            a.onReceiveOrNull { value ->
                if (value == null)
                    "Channel 'a' is closed"
                else
                    "a -> $value"
            }

            b.onReceiveOrNull { value ->
                if (value == null)
                    "Channel 'b' is closed"
                else
                    "b -> $value"
            }
        }

fun do35() {
    runBlocking {
        println("do35()")
        val fizz = fizz(coroutineContext)
        val buzz = buzz(coroutineContext)
        repeat(7)
        {
            selectFizzBuzz(fizz, buzz)
        }

        coroutineContext.cancelChildren()
    }
}

suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
    select<Unit> {
        fizz.onReceive { value ->
            println("fizz -> $value")
        }
        buzz.onReceive { value ->
            println("buzz -> $value")
        }
    }
}

fun fizz(context: CoroutineContext) = produce<String> {
    while (true) {
        delay(300)
        send("Fizz")
    }
}

fun buzz(context: CoroutineContext) = produce<String> {
    while (true) {
        delay(500)
        send("Buzz")
    }
}

fun do34() {
    runBlocking {
        val counter = counterAction()

        println("do34()")
        massiveRun(CommonPool)
        {
            counter.send(IncCounter)
        }

        println("here1")

        val response = CompletableDeferred<Int>()
        counter.send(GetCounter(response))
        println("Counter = ${response.await()}")
        counter.close()

        println("DONE!")
    }
}

fun counterAction() = actor<CounterMsg> {
    var counter = 0
    for (msg in channel) {
        when (msg) {
            is IncCounter -> counter++
            is GetCounter -> msg.response.complete(counter)
        }
    }
}


val mutex = Mutex()
var cnt2 = 0
fun do33() {
    runBlocking {
        massiveRun(CommonPool)
        {
            mutex.withLock {
                cnt2++
            }
        }

        println("Counter = $cnt2")
    }
}

var cnt1 = 0
val counterContext = newSingleThreadContext("XX")
fun do32() {
    println("do32()")

    runBlocking {
        massiveRun(CommonPool)
        {
            withContext(counterContext)
            {
                cnt1++
            }
        }

        println("Counter = $cnt1")
    }
}


var counter = AtomicInteger(0)
fun do31() {
    runBlocking {
        massiveRun(counterContext)
        {
            //            counter.incrementAndGet()
            cnt1++
        }

        println("Counter = ${cnt1}")
    }
}

suspend fun massiveRun(context: CoroutineContext, action: suspend () -> Unit) {
    val n = 1000
    val k = 1000
    val time = measureTimeMillis {
        val jobs = List(n)
        {
            launch(context) {
                repeat(k)
                {
                    action()
                }
            }
        }

        jobs.forEach { it.join() }
    }

    println("Completed ${n * k} actions in $time ms")
}


fun do30() {
    runBlocking {
        val table = Channel<Ball>()
        launch(kotlin.coroutines.experimental.coroutineContext) { player("ping", table) }
        launch(kotlin.coroutines.experimental.coroutineContext) { player("pong", table) }
        table.send(Ball(0))
        delay(1000)

        kotlin.coroutines.experimental.coroutineContext.cancelChildren()
    }
}

suspend fun player(name: String, table: Channel<Ball>) {
    for (ball in table) {
        ball.hits++
        println("$name $ball")
        delay(300)
        table.send(ball)
    }
}

fun do29() {
    runBlocking {
        val ticketChannel = ticker(delay = 100, initialDelay = 0, mode = TickerMode.FIXED_DELAY)
        var e = withTimeoutOrNull(1)
        {
            ticketChannel.receive()
        }
        println("element is $e")
        delay(150)

        e = withTimeoutOrNull(1)
        {
            ticketChannel.receive()
        }
        println("element is $e")


        delay(150)
        e = withTimeoutOrNull(1)
        {
            ticketChannel.receive()
        }
        println("element is $e")

        delay(150)
        e = withTimeoutOrNull(1)
        {
            ticketChannel.receive()
        }
        println("element is $e")

        delay(50)
        e = withTimeoutOrNull(1)
        {
            ticketChannel.receive()
        }
        println("element is $e")
    }
}

fun do28() {
    println("do28()")

    runBlocking<Unit> {
        val tickerChannel = ticker(delay = 100, initialDelay = 0)
        var nextElement = withTimeoutOrNull(1) {
            tickerChannel.receive()
        }
        println("Initial element is available immediately: $nextElement")

        nextElement = withTimeoutOrNull(50)
        {
            tickerChannel.receive()
        }
        println("Next element is not ready in 50ms: $nextElement")

        nextElement = withTimeoutOrNull(60)
        {
            tickerChannel.receive()
        }
        println("Next element is ready in 100ms: $nextElement")

        println("Consumer pauses for 150ms")
        delay(150)
        nextElement = withTimeoutOrNull(1)
        {
            tickerChannel.receive()
        }
        println("Next element is availabl eimmediately after large consumer delay: $nextElement}")

        nextElement = withTimeoutOrNull(60)
        {
            tickerChannel.receive()
        }
        println("Next element is ready in 50s after consmer pause in 150m s: $nextElement")

        tickerChannel.cancel()

        println("Done!")

    }
}

fun do27() {
    println("27()")
    runBlocking {
        val channel = Channel<Int>(4)
        val sender = launch(kotlin.coroutines.experimental.coroutineContext) {
            repeat(10)
            {
                println("Sending $it")
                channel.send(it)
            }
        }

        delay(1000L)
        sender.cancel()
        println("Done!")
    }
}

fun do26() {
    println("do26()")

    runBlocking {
        val numbers = produceNumbers()
        val squares = square(numbers)
        for (i in 1..5)
            println(squares.receive())

        println("Done!")
        squares.cancel()
        numbers.cancel()
    }
}

fun produceNumbers() = produce<Int> {
    var x = 1
    while (true)
        send(x++)
}

fun square(numbers: ReceiveChannel<Int>) = produce<Int> {
    for (x in numbers)
        send(x * x)
}

fun do25() {
    println("25()")

    runBlocking<Unit> {
        val squares = produceSquares()

        for (y in squares)
            println("squares->" + y)

        println("Done!")
    }
}

fun produceSquares(): ReceiveChannel<Int> {
    return produce<Int> {
        for (x in 1..5)
            send(x)
    }
}

fun do24() {
    println("do24()")

    runBlocking {
        val channel = Channel<Int>()
        launch {
            for (x in 1..5) channel.send(x)
            channel.close()
            for (x in 6..9)
                channel.send(x)
        }

        for (y in channel)
            println(y)

        println("Done!")
    }
}

fun do23() {
    runBlocking {
        val channels = Channel<Int>()
        val job = launch {
            delay(1000L)
            for (x in 1..5) {
                println("send $x")
                channels.send(x)
            }
        }


        repeat(5)
        {
            println("receive" + channels.receive())
        }

        println("Done!")
    }
}

fun do22() {
    runBlocking {
        val handler = CoroutineExceptionHandler { _, exception ->
            println("Caught $exception")
        }

        val job = launch(handler) {
            throw AssertionError()
        }

        val deferred = async(handler) {
            throw ArithmeticException()
        }

        joinAll(job, deferred)
    }
}

fun do21() {
    runBlocking {
        val job = launch {
            println("Throwing exception from launch")
//            throw IndexOutOfBoundsException()
        }

        job.join()
        println("joined failed job")

        val deferred = async {
            println("Throwing exception from async")
            throw ArithmeticException()
        }

        try {
//            delay(5000)
            deferred.join()
//            deferred.await()
            println("Unreached")
        } catch (e: Exception) {
            println("Caught: $e")
        }


    }
}

fun do20() {
    runBlocking {
        val job = Job()
        val list = List(10)
        { i ->
            launch(kotlin.coroutines.experimental.coroutineContext, parent = job) {
                delay(600)
                log("Coroutine $i is done")
            }
        }

        println("Launched ${list.size} coroutines")
        delay(500L)
//        println("Cancelling the job")
//        job.cancelAndJoin()

        list.forEach {
            it.join()
        }

        println("1")
        println("2")
    }
}

fun do19() {
    runBlocking(CoroutineName("main")) {
        log("Started main coroutine")
        val v1 = async(CoroutineName("v1coroutine")) {
            delay(500)
            log("Computing v1")
            252
        }

        val v2 = async {
            delay(1000)
            log("Computing v2")
            6
        }

        log("The answer for v1 / v2 = ${v1.await() / v2.await()}")
    }
}

fun do18() {
    runBlocking<Unit> {
        // launch a coroutine to process some kind of incoming request
        val request = launch {
            repeat(3) { i ->
                // launch a few children jobs
                launch(coroutineContext) {
                    delay((i + 1) * 200L) // variable delay 200ms, 400ms, 600ms
                    println("Coroutine $i is done")
                }
            }
            println("request: I'm done and I don't explicitly join my children that are still active")
        }
        request.join() // wait for completion of the request, including all its children
        println("Now processing of the request is complete")
    }
}

fun do17() {
    runBlocking<Unit> {
        val request = launch {
            val jobs = arrayListOf<Job>()
            repeat(3)
            { i ->
                jobs += launch(kotlin.coroutines.experimental.coroutineContext) {
                    delay((i + 1) * 200)
                    println("Coroutine $i is done")
                }
            }

            jobs.forEach {
                //                it.join()
            }
            println("request: I'm done and I dont explicity join my children join that are still active")
        }

        request.join()
        println("Now processing of the request is complete")
    }
}

fun do16() {
    runBlocking {
        val request = launch {
            val job1 = launch {
                println("job1 I have my own context")
                delay(1000L)
                println("job1 I am not affected by cancel")
            }
            val job2 = launch(kotlin.coroutines.experimental.coroutineContext) {
                delay(100)
                println("job2 I am a child of request coroutine")
                delay(1000L)
                println("job2 I will not execute this line")
            }

            job1.join()
            job2.join()
        }

        delay(500)
        request.cancel()
        request.join()
        delay(1000)
        println("main: who has survived request cancellation?")
    }
}

fun do15() {
    runBlocking {
        println("My job is ${kotlin.coroutines.experimental.coroutineContext[Job]}")
    }
}

fun do14() {
    runBlocking {
        val a = async(kotlin.coroutines.experimental.coroutineContext) { delay(1); log("I'm computing a piece of the answer"); 6 }
        val b = async(kotlin.coroutines.experimental.coroutineContext) { log("I'm computing another piece of the answer"); 7 }

        log("The answer is ${a.await() * b.await()}")
    }

}

fun do13() {
    runBlocking {
        val jobs = arrayListOf<Job>()
        jobs += launch(Unconfined) {
            println("1. thread=${Thread.currentThread().name}")
            delay(2500)
            println("2. thread=${Thread.currentThread().name}")
        }

        jobs += launch(kotlin.coroutines.experimental.coroutineContext) {
            println("3. thread=${Thread.currentThread().name}")
            delay(1000L)
            println("4. thread=${Thread.currentThread().name}")
        }

        jobs.forEach {
            it.join()
        }
    }
}

fun do12() {
    runBlocking {
        val jobs = arrayListOf<Job>()
        jobs += launch(Unconfined) { println("Unconfined: thread ${Thread.currentThread().name}") }
        jobs += launch(kotlin.coroutines.experimental.coroutineContext) { println("coroutineContext: thread ${Thread.currentThread().name}") }
        jobs += launch(CommonPool) { println("CommonPool: thead=${Thread.currentThread().name}") }
        jobs += launch(newSingleThreadContext("MyOwnThread")) { println("newSTC: thread=${Thread.currentThread().name}") }

        jobs.forEach {
            it.join()
        }

    }
}

fun do11() {
    val time = measureTimeMillis {
        val one = somethingOneAsync()
        val two = somethingTwoAsync()

        runBlocking {
            println("The answer is ${one.await() + two.await()}")
        }
    }

    println("Completed is $time ms")
}

fun somethingTwoAsync() = async {
    doSomethingTwo()
}

fun somethingOneAsync() = async {
    doSomethingOne()
}

fun do10() {
    runBlocking {
        val time = measureTimeMillis {
            val one = async(start = CoroutineStart.LAZY) { doSomethingOne() }
            val two = async(start = CoroutineStart.LAZY) { doSomethingTwo() }
            println("1")
            println("one=${one.await()}")
            println("2")
            two.start()
            println("3")
            println("two=${two.await()}")
        }
    }
}

fun do9() {
    runBlocking {
        val time = measureTimeMillis {
            val one = async { doSomethingOne() }
            val two = async { doSomethingTwo() }
            println("one=$one, two=$two")
            two.cancel()
            println("one=$one, two=$two")
            println("The answer is ${one.await() + two.await()}")
        }

        println("completed in $time ms")
    }
}

fun do8() {
    runBlocking {
        val time = measureTimeMillis {
            val one = doSomethingOne()
            val two = doSomethingTwo()
            println("result=${one + two}")
        }

        println("run time = $time")
    }
}

suspend fun doSomethingTwo(): Int {
    delay(2000L)
    return 200
}

suspend fun doSomethingOne(): Int {
    delay(1000L)
    return 100
}

fun do7() {
    runBlocking<Unit> {
        val r = withTimeoutOrNull(1300L)
        {
            repeat(100)
            { i ->
                println("I'm sleeping $i")
                delay(1L)
            }

            "Done"
        }

        println("r=$r")
        println("cancel and end")
    }
}

fun do6() {
    runBlocking<Unit> {
        val job = launch {
            try {
                repeat(1000) { i ->
                    println("I'm sleeping $i ...")
                    delay(500L)
                }
            } finally {
                withContext(NonCancellable) {
                    println("I'm running finally")
                    delay(1000L)
                    println("And I've just delayed for 1 sec because I'm non-cancellable")
                }
            }
        }
        delay(1300L) // delay a bit
        println("main: I'm tired of waiting!")
        job.cancelAndJoin() // cancels the job and waits for its completion
        println("main: Now I can quit.")
    }
}

fun do5() {
    runBlocking {
        val job = launch {
            try {
                repeat(1000) { i ->
                    println("I'm sleeping $i ...")
                    delay(500L)
                }
            } catch (e: Exception) {
                delay(1500L)
                println("I'm running catch $e")
            } finally {
                delay(1500L)
                println("I'm running finally")
            }

        }
        delay(1300L) // delay a bit
        println("main: I'm tired of waiting!")
        job.cancelAndJoin() // cancels the job and waits for its completion
        println("main: Now I can quit.")
    }
}


fun do4() {
    runBlocking<Unit> {
        val startTime = System.currentTimeMillis()
        val job = launch {
            var nextPrintTime = startTime
            var i = 0
            while (isActive) { // cancellable computation loop
                // print a message twice a second
                if (System.currentTimeMillis() >= nextPrintTime) {
                    println("I'm sleeping ${i++} ...")
                    nextPrintTime += 500L
                }
            }
        }
        delay(1300L) // delay a bit
        println("main: I'm tired of waiting!")
        job.cancelAndJoin() // cancels the job and waits for its completion
        println("main: Now I can quit.")
    }
}


fun do3() {
    runBlocking<Unit> {
        val job = launch {
            repeat(1_000)
            {
                println("I'm sleeping $it")
                delay(5L)
            }
        }

        delay(500L)
        println("main: I'm tired of waiting!")
        job.cancel()
        job.join()
        println("main: Now I can quit.")
    } //runBlocking
}


private fun do2() {
    runBlocking<Unit> {

        launch {
            repeat(1000) { i ->
                println("I'm sleeping $i ...")
                delay(500L)
            }
        }
        delay(1300L) // just quit after delay
    }
}

private fun do1() {
    runBlocking<Unit> {
        val jobs = List(100_000) {
            launch {
                delay(1000L)
                println(".")
            }
        }

        jobs.forEach {
            it.join()
        }
    }
}

private suspend fun doWork() {
    delay(1000L)
    println("World")
}